package org.jeecg.modules.cascv.kafka.listener;

import cn.hutool.core.io.FileUtil;
import com.alibaba.fastjson.JSONObject;
import com.google.common.collect.EvictingQueue;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.caffinitas.ohc.Eviction;
import org.caffinitas.ohc.OHCache;
import org.caffinitas.ohc.OHCacheBuilder;
import org.jeecg.modules.cascv.image.service.impl.MergeImageService;
import org.jeecg.modules.cascv.image.service.impl.MergeVideoService;
import org.jeecg.modules.cascv.image.util.CacheJsonSerializer;
import org.jeecg.modules.cascv.image.util.CacheStringSerializer;
import org.jeecg.modules.cascv.image.util.LaptopConstant;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

import java.util.concurrent.atomic.AtomicLong;

@Slf4j
//@Component
public class KafkaListeners {

    @Autowired
    private MergeImageService mergeImageService;
    @Autowired
    private MergeVideoService mergeVideoService;

  //  @KafkaListener(topics = {"image"}, groupId = "image222")
    public void imageListener(ConsumerRecord<?, ?> record) {
        String topic = record.topic();
        Object value = record.value();
        if (value == null) {
            log.error("kafka消息内容为空！");
            return;
        }
        String jsonStr = value.toString();
        JSONObject jsonObject = JSONObject.parseObject(jsonStr);

        mergeImageService.mergeImage(jsonObject, null);

    }

//    @KafkaListener(topics = {"video"}, groupId = "video")
    public void videoListener(ConsumerRecord<?, ?> record) {
        String topic = record.topic();
        Object value = record.value();
        if (value == null) {
            log.error("kafka消息内容为空！");
            return;
        }
        String jsonStr = value.toString();
        JSONObject jsonObject = JSONObject.parseObject(jsonStr);

        mergeVideoService.merge(jsonObject, null, null);

//        EvictingQueue<JSONObject> evictingQueue = LaptopConstant.videoData.get(tableName);
//        if(evictingQueue == null){
//            synchronized (LaptopConstant.videoData){
//                evictingQueue = LaptopConstant.videoData.get(tableName);
//                if(evictingQueue == null){
//                    evictingQueue = EvictingQueue.create(10000);
//                    LaptopConstant.videoData.put(tableName, evictingQueue);
//                    LaptopConstant.videoIndex.put(tableName, new AtomicLong(0));
//                }
//            }
//        }
//
//        long l = LaptopConstant.videoIndex.get(tableName).addAndGet(1);
//        evictingQueue.add(jsonObject);

    }

}